package com.ww.spark.core

import org.apache.spark.{SparkConf, SparkContext}

import scala.util.control.Breaks.{break, breakable}

object RDDtry05_api_over {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("try05")
    val sc = new SparkContext(conf)
    val file = sc.textFile("data/tqdata")

    implicit val abc = new Ordering[(Int, Int)] {
      override def compare(x: (Int, Int), y: (Int, Int)) = y._2.compareTo(x._2)
    }

    /*println("\n 同月份中温度最高的两天======================================================================================")
    //通过line 得到 (年月,天,温度)
    val tup = file.map(line => line.split("\t")).map(arr => {
      val arrs = arr(0).split("-")
      (s"${arrs(0)}-${arrs(1)}", arrs(2).toInt, arr(1).toInt)
    })
    //通过年月group
    val grouped = tup.map(t => ((t._1), (t._2, t._3))).groupByKey()
    //得到保留group中温度最高的一天
    val res = grouped.mapValues(arr => {
      val map = new mutable.HashMap[Int, Int]
      //(x,x)
      arr.foreach(e => {
        if (map.getOrElse(e._1, -100) < e._2) {
          map.put(e._1, e._2)
        }
      })
      //排序
      val tuples = map.toList.sorted(new Ordering[(Int, Int)] {
        override def compare(x: (Int, Int), y: (Int, Int)): Int = y._2.compareTo(x._2)
      })
      tuples
    })

    //输出
    res.foreach(println)
    println("\n 同月份中温度最高的两天 去掉map======================================================================================")
    val tup1 = file.map(line => line.split("\t")).map(arr => {
      val arrs = arr(0).split("-")
      ((arrs(0), arrs(1), arrs(2).toInt), arr(1).toInt)
    })
    //找到最大值
    val reduced = tup1.reduceByKey((o: Int, n: Int) => {
      if (n > o) n else o
    })

    //reduced.foreach(println)
    //    ((1970,8,8),32)
    //    ((2019,6,2),31)
    //    ((2018,4,23),22)
    //    ((2019,6,1),39)
    //    ((2018,3,11),18)
    //    ((2019,5,21),33)
    //    ((1970,8,23),23)
    val maped = reduced.map(t => ((t._1._1, t._1._2), (t._1._3, t._2)))
    //maped.foreach(println)
    //    ((1970,8),(8,32))
    //    ((2019,6),(2,31))
    //    ((2018,4),(23,22))
    //    ((2019,6),(1,39))
    //    ((2018,3),(11,18))
    //    ((2019,5),(21,33))
    //    ((1970,8),(23,23))
    val grouped1 = maped.groupByKey()
    //grouped1.foreach(println)
    //    ((1970,8),CompactBuffer((8,32), (23,23)))
    //    ((2018,3),CompactBuffer((11,18)))
    //    ((2018,4),CompactBuffer((23,22)))
    //    ((2019,6),CompactBuffer((2,31), (1,39)))
    //    ((2019,5),CompactBuffer((21,33)))

    val res1 = grouped1.mapValues(g => {
      g.toList.sorted(new Ordering[(Int, Int)] {
        override def compare(x: (Int, Int), y: (Int, Int)): Int = y._2.compareTo(x._2)
      })
    })
    res1.foreach(println)
    //Thread.sleep(1000000)
    println("\n 同月份中温度最高的两天 ======================================================================================")

    val tup2 = file.map(line => line.split("\t")).map(arr => {
      val arrs = arr(0).split("-")
      (arrs(0), arrs(1), arrs(2).toInt, arr(1).toInt)
    })
    val sorted = tup2.sortBy(t => (t._1, t._2, t._4), false)
    sorted.foreach(println)
    sorted.map(t => ((t._1, t._2), (t._3, t._4))).groupByKey().foreach(println)*/
    println("\n 同月份中温度最高的两天 最终版======================================================================================")

    //用combine 在 每个stage上先合并数据
    val tup3 = file.map(line => line.split("\t")).map(arr => {
      val arrs = arr(0).split("-")
      (arrs(0), arrs(1), arrs(2).toInt, arr(1).toInt)
    })
    //先转换成Tuple2
    val tup4 = tup3.map(t4 => ((t4._1, t4._2), (t4._3, t4._4)))
    //combin
    val com = tup4.combineByKey(
      //第一条记录
      (value: (Int, Int)) => {
        Array(value, (0, 0), (0, 0))
      },
      //第二条记录
      (o: Array[(Int, Int)], n: (Int, Int)) => {
        //1.如果日期相同,o中替换成温度较大的 flag = 1 ,返回 o
        //2.如果日期不同,放在最后一个格子里,然后排序 0,放回 o
        var flag = 0
        breakable {
          for (i <- o.indices) {
            if (o(i)._1 == n._1) {
              if (o(i)._2 < n._2) {
                //1.如果日期相同,取温度较大的
                o(i) = n
              }
              flag = 1
              break
            }
          }
        }
        if (flag == 0) {
          o(o.length - 1) = n
        }
        scala.util.Sorting.quickSort(o)
        o
      },
      //合并
      (v1: Array[(Int, Int)], v2: Array[(Int, Int)]) => {
        //union 然后返回前2
        val tuples = v1.union(v2).sorted
        tuples.take(2)
      }
    )
    com.map(e => (e._1, e._2.toList)).foreach(println)

    //Thread.sleep(10000000)

    println("\n try ======================================================================================")
    val tuples = Array((0, 0), (0, 0), (0, 0))
    breakable {
      for (i <- tuples.indices) {
        if (i == 2) break
        println(i)
      }
    }
  }
}
